package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.elasticsearch.common.CheckedFunction;

import java.util.ArrayList;
import java.util.List;

public class TestProcessFunction extends ProcessFunction<Integer, Integer> implements CheckpointedFunction {

    private ListState<Integer> listState;
    private List<Integer> list =new ArrayList<>();

    @Override
    public void processElement(Integer value, ProcessFunction<Integer, Integer>.Context ctx, Collector<Integer> out) throws Exception {
        //把来的每一条数据都缓存到列表中
        list.add(value);
        //判断如果达到阈值，就批量写入
        if (list.size() == 5) {
            String s = "buffers:";
            //用打印到控制台模拟写入外部系统
            for (Integer i : list) {
                s=s+i+",";
            }
            System.out.println(s);
            System.out.println("====================输出完毕====================");
            list.clear();
        }
        out.collect(value);
    }


    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        //清空状态，保证状态跟这里的bufferedElements完全一样
        listState.clear();

        //对状态进行持久化，复制缓存的列表到列表状态
        for (Integer i : list) {
            listState.add(i);
        }
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Integer> listStateDescriptor = new ListStateDescriptor<>("listStateDescriptor", Integer.class );
        listState = context.getOperatorStateStore().getListState( listStateDescriptor );

    }
}


